%%-*- mode: erlang -*-
%% emqx_auth_pgsl config mapping

{mapping, "kafka.brokers", "quec_emq_kafka.bootstrap_brokers", [
  {default, "127.0.0.1:9092"},
  {datatype, string}
]}.

{translation, "quec_emq_kafka.bootstrap_brokers", fun(Conf) ->
  Brokers = cuttlefish:conf_get("kafka.brokers", Conf),
  BrokerList = string:tokens(Brokers, ","),
  BrokerParser = fun(Broker) ->
    [Host|[PortStr|Rest]] = string:tokens(Broker, ":"),
    Port = list_to_integer(PortStr),
    {Host, Port}
  end,
  lists:map(BrokerParser, BrokerList)
  end}.

{mapping, "kafka.config.instrument", "quec_emq_kafka.config", [
  {default, false},
  {datatype, {enum, [true, false]}}
]}.

{translation, "quec_emq_kafka.config", fun(Conf) ->
  Instrument = cuttlefish:conf_get("kafka.config.instrument", Conf),
  [{instrument, Instrument}]
  end}.

{mapping, "kafka.points.topic", "quec_emq_kafka.points", [
  {default, "device-data-points"},
  {datatype, string}
]}.

{mapping, "kafka.points.partition.strategy", "quec_emq_kafka.points", [
  {default, "strict_round_robin"},
  {datatype, string}
]}.

{mapping, "kafka.points.partition.workers", "quec_emq_kafka.points", [
  {default, 8},
  {datatype, integer}
]}.

{translation, "quec_emq_kafka.points", fun(Conf) ->
  Topic = cuttlefish:conf_get("kafka.points.topic", Conf),
  PartitionStrategy = cuttlefish:conf_get("kafka.points.partition.strategy", Conf),
  PartitionWorkers = cuttlefish:conf_get("kafka.points.partition.workers", Conf),
  [{topic, list_to_binary(Topic)}, {partition_strategy, list_to_atom(PartitionStrategy)}, {partition_workers, PartitionWorkers}]
  end}.

{mapping, "kafka.status.topic", "quec_emq_kafka.status", [
  {default, "device-data-points"},
  {datatype, string}
]}.

{mapping, "kafka.status.partition.strategy", "quec_emq_kafka.status", [
  {default, "strict_round_robin"},
  {datatype, string}
]}.

{mapping, "kafka.status.partition.workers", "quec_emq_kafka.status", [
  {default, 8},
  {datatype, integer}
]}.

{translation, "quec_emq_kafka.status", fun(Conf) ->
  Topic = cuttlefish:conf_get("kafka.status.topic", Conf),
  PartitionStrategy = cuttlefish:conf_get("kafka.status.partition.strategy", Conf),
  PartitionWorkers = cuttlefish:conf_get("kafka.status.partition.workers", Conf),
  [{topic, list_to_binary(Topic)}, {partition_strategy, list_to_atom(PartitionStrategy)}, {partition_workers, PartitionWorkers}]
  end}.

